package com.swak.reactivex.p2p.exchange;

import static com.swak.Constants.CLIENT_SIDE;
import static com.swak.Constants.SIDE;

import java.util.Map;
import java.util.concurrent.CompletionStage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.reactivex.p2p.Peer;
import com.swak.reactivex.p2p.PeerID;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeClient;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.Message;
import com.swak.registry.URL;

/**
 * 管理 ExchangeChannel.
 * 只需要保持一个连接即可.
 * 
 * @author 618lf
 */
@Deprecated
public class ExchangeGroupHandler implements ExchangeHandler {
	private static final Logger logger = LoggerFactory.getLogger(Peer.class);

	private final PeerID peerID;
	private final ExchangeGroup group;
	private final ExchangeHandler delegate;
	private final Map<URL, ExchangeClient> clients;
	private final ExchangeChannels channels;

	public ExchangeGroupHandler(PeerID peerID, ExchangeGroup group, ExchangeHandler delegate,
			Map<URL, ExchangeClient> clients, ExchangeChannels channels) {
		this.peerID = peerID;
		this.group = group;
		this.delegate = delegate;
		this.clients = clients;
		this.channels = channels;
	}

	/**
	 * 两端有可能同时发起连接的情况，为了保证只留一条连接：
	 * 如果连接建立成功时发现 channels 中已经包含了连接，则主动断开连接；
	 * 但可能解决不了所有的问题，因为也有可能都作为客户端主动的断开连接；
	 * 但是至少可以解决大部分的问题；后续的joinTask 会再次发起连接通知；
	 */
	@Override
	public void connected(Channel channel) throws RemotingException {
		String side = (String) channel.getAttribute(SIDE);
		if (CLIENT_SIDE.equals(side)) {
			PeerID peerID = PeerID.remotePeerID(channel.url());
			if (this.channels.containsKey(peerID)) {
				this.group.disconnect(channel.url());
				return;
			}
			this.channels.put(peerID, (ExchangeChannel) channel);
			this.delegate.connected(channel);
			this.sendLocalPeeId(channel);
		}
	}

	private void sendLocalPeeId(Channel channel) throws RemotingException {
		channel.send(Message.forMessage().setData(this.peerID));
		if (logger.isDebugEnabled()) {
			logger.debug("Send: " + this.peerID);
		}
	}

	private void recvRemotePeeId(Channel channel, PeerID peerID) throws RemotingException {
		if (logger.isDebugEnabled()) {
			logger.debug("Recv: " + peerID);
		}
		channel.setAttribute(Constants.Peer_Id, peerID);
		this.channels.put(peerID, (ExchangeChannel) channel);
		this.delegate.connected(channel);
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		String side = (String) channel.getAttribute(SIDE);
		if (CLIENT_SIDE.equals(side)) {
			PeerID peerID = PeerID.remotePeerID(channel.url());
			this.channels.remove(peerID);
		} else {
			PeerID peerID = (PeerID) channel.getAttribute(Constants.Peer_Id);
			this.channels.remove(peerID);
		}
		this.clients.remove(channel.url());
		this.delegate.disconnected(channel);
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		this.delegate.sent(channel, message);
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		Object msg = ((Message) message).getData();
		if (msg instanceof PeerID) {
			this.recvRemotePeeId(channel, (PeerID) msg);
			return;
		}
		this.delegate.received(channel, message);
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		this.delegate.caught(channel, exception);
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		return this.delegate.reply(channel, message);
	}
}
